package com.calabar.phm.etl.driver.spark;

import com.calabar.phm.etl.driver.spark.exception.DriverException;
import com.calabar.phm.etl.driver.spark.exception.ExceptionCodeEnum;

import java.io.FileReader;
import java.io.InputStream;
import java.sql.*;
import java.util.Properties;

/**
 * @Author zmc <mingcheng.zhang@cdcalabar.com>
 * @Date 17-6-9 下午4:17
 * @Version v1.0
 * @Des 记录spark任务运行状态
 */
public class Recorder {
    /**
     * 数据库连接
     */
    private Connection conn;

    public Recorder(String url, String userName, String userPasswd) throws DriverException {
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection(url, userName, userPasswd);
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(),
                    "初始化mysql连接错误，请检查数据库配置！", e);
        }
    }

    /**
     * 插入驱动运行状态（初始化）
     *
     * @param jobId           任务标识
     * @param subJobId        子任务标识
     * @param driverStateCode 驱动状态
     * @throws DriverException
     */
    public void insertJobState(String jobId, String subJobId, int driverStateCode) throws DriverException {
        try {
            String sql = "insert into etl_job(job_id,subjob_id,driver_state) values (?,?,?);";
            Connection conn = getConn();
            PreparedStatement ps = conn.prepareStatement(sql);
            ps.setString(1, jobId);
            ps.setString(2, subJobId);
            ps.setInt(3, driverStateCode);
            ps.executeUpdate();
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(), "记录驱动状态出现异常！", e);
        }
    }

    /**
     * 更新驱动运行状态
     *
     * @param jobId           任务标识
     * @param subJobId        子任务标识
     * @param driverStateCode 驱动状态
     * @throws DriverException
     */
    public void updateJobState(String jobId, String subJobId, int driverStateCode) throws DriverException {
        try {
            String sql = "update etl_job set driver_state=? where job_id=? and subjob_id=?;";
            Connection conn = getConn();
            PreparedStatement ps = conn.prepareStatement(sql);
            ps.setInt(1, driverStateCode);
            ps.setString(2, jobId);
            ps.setString(3, subJobId);
            ps.executeUpdate();
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(), "记录驱动状态出现异常！", e);
        }
    }

    /**
     * 记录spark任务的appid
     *
     * @param jobId    任务标识
     * @param subJobId 子任务标识
     * @param appId    appid
     * @throws DriverException
     */
    public void updateAppID(String jobId, String subJobId, String appId) throws DriverException {
        try {
            String sql = "update etl_job set app_id=? where job_id=? and subjob_id=?;";
            Connection conn = getConn();
            PreparedStatement ps = conn.prepareStatement(sql);
            ps.setString(1, appId);
            ps.setString(2, jobId);
            ps.setString(3, subJobId);
            ps.executeUpdate();
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(), "记录appid出现异常！", e);
        }
    }

    /**
     * 检查任务是否重复提交
     *
     * @param jobId    任务标识
     * @param subJobId 子任务标识
     * @return
     * @throws DriverException
     */
    public boolean isSubmitted(String jobId, String subJobId) throws DriverException {
        int count = 0;
        try {
            String sql = "select count(*) from etl_job where job_id=? and subjob_id=?;";
            Connection conn = getConn();
            PreparedStatement ps = conn.prepareStatement(sql);
            ps.setString(1, jobId);
            ps.setString(2, subJobId);
            ResultSet resultSet = ps.executeQuery();
            if (resultSet.next()) {
                count = resultSet.getInt(1);
            }
            return count != 0 ? true : false;
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(), "验证任务是否重复提交出现异常！", e);
        }
    }

    /**
     * 记录任务运行产生的错误日志
     * @param jobId
     * @param subJobId
     * @param driverStateCode
     * @param code
     * @param errLog
     * @param operatorUnique
     * @throws DriverException
     */
    public void recordJobStateAndErrorLog(String jobId, String subJobId, int driverStateCode,
                                          int code, String errLog, String operatorUnique)
            throws DriverException {
        try {
            String sql = "update etl_job set message=?,driver_state=?,operator_unique=?,code=? where job_id=? and subjob_id=?";
            Connection conn = getConn();
            PreparedStatement ps = conn.prepareStatement(sql);
            ps.setString(1, errLog);
            ps.setInt(2, driverStateCode);
            ps.setString(3, operatorUnique);
            ps.setInt(4, code);
            ps.setString(5, jobId);
            ps.setString(6, subJobId);
            ps.executeUpdate();
        } catch (Exception e) {
            throw new DriverException(ExceptionCodeEnum.DRIVER_INIT_EXCEPTION.get_code(), "记录驱动运行错误信息出现异常！", e);
        }
    }

    public Connection getConn() {
        return conn;
    }

    public void close() throws SQLException {
        if (conn != null) {
            conn.close();
        }
    }
}
